package com.mlamp.me;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.SqlDialect;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.catalog.hive.HiveCatalog;

/**
 * Desc
 */
public class HiveDemoCreateTwo {
    public static void main(String[] args){
        EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().build();
        TableEnvironment tableEnv = TableEnvironment.create(settings);

        String name            = "myhive";
        String defaultDatabase = "hive_tmp";
        String hiveConfDir     = "./src/main/resources";
        String version         = "2.3.4";

        HiveCatalog hive = new HiveCatalog(name, defaultDatabase, hiveConfDir,version);
        //注册catalog  hive_tmp.log_hive
        tableEnv.registerCatalog("hive_tmp", hive);
        //使用注册的catalog
        tableEnv.useCatalog("hive_tmp");

        tableEnv.getConfig().setSqlDialect(SqlDialect.HIVE);
        /**
         * 不能使用 if not exists
         */
        //向Hive表中写入数据
        TableResult result = null;
        try {
            String createSql = "CREATE TABLE  dm_info_test_df_nopart " +
                    "( " +
                    "    group_code string comment '维度编码'," +
                    "    group_name string comment '维度名称'," +
                    "    prov_code  string comment '省区编码'," +
                    "    prov_name  string comment '省区名称'" +
                    ") ";
            System.out.println(createSql);
            result = tableEnv.executeSql(createSql);
            result.collect().forEachRemaining(ele -> System.out.println("out==>> "+ ele.toString()));
        } catch (Exception e) {
            System.out.println(e.getMessage());
        }

        String insertSql = "insert into dm_info_test_df_nopart  " +
                "values"
                +"('组一','北京','bj','beijing'),"
                +"('组二','深圳','sz','shenzhen')"
                ;
        tableEnv.executeSql(insertSql);

        String selectSQL = "select * from dm_info_test_df_nopart";
        TableResult selectResult = tableEnv.executeSql(selectSQL);
        selectResult.collect().forEachRemaining(ele -> System.out.println("select==>> "+ ele.toString()));


//        System.out.println(result.getJobClient().get().getJobStatus());
    }
}